package app

import com.cw.realtime.common.base.FlinkRunner.{SqlRunnerConfig, runSql}
import com.cw.realtime.common.constant.Constant.{DWS_TRAFFIC_KEYWORD_PAGE_VIEW_WINDOW, TOPIC_DWD_TRAFFIC_PAGE}
import com.cw.realtime.common.util.{IkUtil, SqlUtil}
import org.apache.flink.table.annotation.{DataTypeHint, FunctionHint}
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.functions.TableFunction
import org.apache.flink.types.Row


@FunctionHint(output = new DataTypeHint("ROW<keyword STRING>"))
class KeywordSplit extends TableFunction[Row] {
  def eval(keywords: String): Unit = {
    IkUtil.ikSplit(keywords).foreach { s =>
      collect(Row.of(s, Int.box(s.length)))
    }
  }
}


object DwsTrafficKeywordPageViewWindow {


  def main(args: Array[String]): Unit = {

    implicit val config = SqlRunnerConfig("DwsTrafficKeywordPageViewWindow", 10021, ckp = true)
    runSql { (env, tableEnv) =>

      // 读取dwd层page日志明细表
      readDwdPageLog(tableEnv)


      // 提取原始关键字数据
      val kwdsTable = extractKeywords(tableEnv)
      tableEnv.createTemporaryView("kwds_table", kwdsTable)
      tableEnv.createTemporaryFunction("kwSplit", classOf[KeywordSplit])


      // 关键字分词
      val keywordTable = splitKeywords(tableEnv)
      tableEnv.createTemporaryView("keyword_table", keywordTable)


      // 关键字分组滚动窗口计数

      val keywordWinAgg = groupWordCountWindow(tableEnv)


      // 创建doris sink
      createDorisSink(tableEnv)

      keywordWinAgg.insertInto("flink_doris_source").execute()


    }


  }


  def readDwdPageLog(tableEnv: StreamTableEnvironment)(implicit conf: SqlRunnerConfig) = {
    tableEnv.executeSql(
      s"""
create table pageInfo (
  `common` map<string, string>,
  `page` map<string, string>,
  ts bigint,
  raw_time as TO_TIMESTAMP_LTZ(ts / 1000, 0),
  WATERMARK FOR raw_time AS raw_time - INTERVAL '2' SECOND
)  ${SqlUtil.getKafkaSourceSql(TOPIC_DWD_TRAFFIC_PAGE)}

        """
    )





  }


  def extractKeywords(tableEnv: StreamTableEnvironment) = {
    tableEnv.sqlQuery(
      """
select
  `page`['item'] keywords,
  ts,
  raw_time
from pageInfo
where
`page`['last_page_id'] = 'search'
and `page`['item_type'] = 'keyword'
and `page`['item'] is not null

        """
    )
  }

  def splitKeywords(tableEnv: StreamTableEnvironment) = {
    tableEnv.sqlQuery(
      """
select
  keywords,
  keyword,
  raw_time
from kwds_table
join LATERAL TABLE(kwSplit(keywords)) ON TRUE
        """
    )
  }


  def groupWordCountWindow(tableEnv: StreamTableEnvironment) = {
    tableEnv.sqlQuery(
      """
SELECT
  cast(TUMBLE_START(raw_time, INTERVAL '10' SECOND) AS STRING) AS stt,
  cast(TUMBLE_END(raw_time, INTERVAL '10' SECOND) AS STRING) AS edt,
  cast(CURRENT_DATE  AS STRING) as cur_date,
  keyword,
  count(*) as keyword_count
FROM keyword_table
GROUP BY
  TUMBLE(raw_time, INTERVAL '10' SECOND ),
  keyword
        """
    )
  }

  def createDorisSink(tableEnv: StreamTableEnvironment) = {
    tableEnv.executeSql(
      s"""
CREATE TABLE flink_doris_source (
     stt STRING,
     edt STRING,
     cur_date STRING,
     keyword STRING,
     keyword_count BIGINT
) ${SqlUtil.getDorisSinkSql(DWS_TRAFFIC_KEYWORD_PAGE_VIEW_WINDOW)}
        """
    )
  }

}

